package com.example.spring_boot_canal.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 要使用show方法 需要放入util包
 */
@Component
public class CanalApplicationRunner implements ApplicationRunner {

    @Autowired
    private CanalConnector connector;

    @Autowired
    private CanalConfig config;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        System.out.println(config.getSubscribe());
        try {
            connector.connect();
            connector.subscribe(config.getSubscribe());
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(config.getBatchSize()); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                try {
                    if (batchId == -1 || size == 0) {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    } else {
                        printEntry(message.getEntries());
                    }
                    connector.ack(batchId); // 提交确认
                } catch (CanalClientException e) {
                    connector.rollback(batchId); // 处理失败, 回滚数据
                    e.printStackTrace();
                }
            }
        }finally {
            connector.disconnect();
        }

    }

    private static void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            CanalEntry.EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                switch (eventType){
                    case INSERT:
                        printColumn(rowData.getAfterColumnsList());
//                        showSql(entry.getHeader() , rowData.getAfterColumnsList());
                        sendInsertEvent(rowData.getAfterColumnsList());
                        break;
                    case UPDATE:
                        sendUpdateEvent(rowData.getBeforeColumnsList(),rowData.getAfterColumnsList());
                        System.out.println("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        System.out.println("-------> after");
                        printColumn(rowData.getAfterColumnsList());
//                        showSql(entry.getHeader() , rowData.getAfterColumnsList());
                        break;
                    case DELETE:
                        sendDeleteEvent(rowData.getBeforeColumnsList());
                        printColumn(rowData.getBeforeColumnsList());
//                        showSql(entry.getHeader() , rowData.getBeforeColumnsList());
                        break;
                }
            }
        }
    }


    private static void sendInsertEvent(List<CanalEntry.Column> columns) {
        MysqlDMLEvent event = new MysqlDMLEvent();
        event.setEventType(CanalEntry.EventType.INSERT);
        List<SqlDMLDetail> details = columns.stream().map(c -> SqlDMLDetail.builder().cloumnName(c.getName()).destinationValue(c.getValue()).build()).collect(Collectors.toList());
        event.setDetails(details);
        System.out.println(event);
    }

    private static void sendUpdateEvent(List<CanalEntry.Column> beforeColumnsList, List<CanalEntry.Column> afterColumnsList) {
        Map<String, Pair<String, String>> map= afterColumnsList.stream().filter(CanalEntry.Column::getUpdated).collect(Collectors.toMap(CanalEntry.Column::getName, c -> Pair.of(null, c.getValue()), (o1, o2) -> o1));

        beforeColumnsList.forEach(c -> {
            Pair<String, String> pair = map.get(c.getName());
            if(pair != null){
                Pair<String, String> newPair = Pair.of(c.getValue(), pair.getRight());
                map.put(c.getName(),newPair);
            }
        });

        List<SqlDMLDetail> details = map.entrySet().stream().map(c -> SqlDMLDetail.builder().cloumnName(c.getKey()).orginValue(c.getValue().getLeft()).destinationValue(c.getValue().getRight()).build()).collect(Collectors.toList());
        MysqlDMLEvent event = new MysqlDMLEvent();
        event.setDetails(details);
        event.setEventType(CanalEntry.EventType.UPDATE);
        System.out.println(event);


    }

    private static void sendDeleteEvent(List<CanalEntry.Column> beforeColumnsList) {
        MysqlDMLEvent event = new MysqlDMLEvent();
        event.setEventType(CanalEntry.EventType.DELETE);
        List<SqlDMLDetail> details = beforeColumnsList.stream().map(c -> SqlDMLDetail.builder().cloumnName(c.getName()).orginValue(c.getValue()).build()).collect(Collectors.toList());
        event.setDetails(details);

        System.out.println(event);
    }


    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

//    private static void showSql(CanalEntry.Header header, List<CanalEntry.Column> columns) {
//        List<String> pkNames = new ArrayList<>();
//        List<String> colNames = new ArrayList<>();
//        List<Object> pkValues = new ArrayList<>();
//        List<Object> colValues = new ArrayList<>();
//        //如果是insert每个字段都更新了，但是有些字段没有值，需要处理
//        for (CanalEntry.Column column : columns) {
//            if (column.getIsKey()) {
//                pkNames.add(column.getName());
//                pkValues.add(column.getValue());
//            } else if (column.getUpdated() && StringUtils.isNotEmpty(column.getValue())){
//                colNames.add(column.getName());
//                colValues.add(column.getValue());
//            }
//        }
//        String sql = "";
//        CanalEntry.EventType eventType = header.getEventType();
//        MysqlSqlTemplate sqlTemplate = new MysqlSqlTemplate();
//        switch (eventType) {
//            case INSERT:
//                sql = sqlTemplate.getInsertSql(header.getSchemaName(), header.getTableName(), pkNames.toArray(new String[]{}), colNames.toArray(new String[]{}), pkValues.toArray(), colValues.toArray());
//                break;
//            case UPDATE:
//                sql = sqlTemplate.getUpdateSql(header.getSchemaName(), header.getTableName(), pkNames.toArray(new String[]{}), colNames.toArray(new String[]{}), pkValues.toArray(), colValues.toArray());
//                break;
//            case DELETE:
//                sql = sqlTemplate.getDeleteSql(header.getSchemaName(), header.getTableName(), pkNames.toArray(new String[]{}), pkValues.toArray());
//        }
//        System.out.println(sql);


}
